/*************************************************************************************
* 	 Copyright (C) 2010 by Information Systems Group, Saarland University  			*
*    http://infosys.cs.uni-saarland.de												*
* 	 																				*
* 	 This file is part of Hadoop++.												 	*
*																					*
*    Hadoop++ is free software: you can redistribute it and/or modify				*
*    it under the terms of the GNU Lesser General Public License as published by	*
*    the Free Software Foundation, either version 3 of the License, or				*
*    (at your option) any later version.											*
*																					*
*    Hadoop++ is distributed in the hope that it will be useful,					*
*    but WITHOUT ANY WARRANTY; without even the implied warranty of					*
*    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the					*
*    GNU Lesser General Public License for more details.							*
*																					*
*    You should have received a copy of the GNU Lesser General Public License		*
*    along with Hadoop++.  If not, see <http://www.gnu.org/licenses/>.				*
*************************************************************************************/
package tpc.benchmarks.joinOnIndex;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.FileSplit;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.RecordReader;

import unisb.cs.core.binary.utils.BinaryUtils;
import unisb.cs.core.index.access.IndexLookupWithLongOffsets;
import unisb.cs.data.tables.TableObjectProxy;

/**
 * This class set the uservisit.sourceIP and ranking.pageRank+UV.adRevenue 
 * as <key, value> pairs for the map function.
 */
public class JoinTaskRecordReader implements RecordReader<Text, Text> {
	
	// split variables
	protected FileSplit fileSplit;
	protected FSDataInputStream in;
	protected long offset = 0;
	protected int key, keyType, keySize;
	
	protected final static int tableDataHeaderLen = 9;
	
	protected String leftTableName;
	protected String rightTableName;	
	protected long rightTableDataSize;

	protected Map<String, List<TableObjectProxy>> leftJoinRecord;
	protected List<TableObjectProxy> joinKeyRecord;
	protected TableObjectProxy currentRightTableObjectProxy;

	protected int leftJoinRecordPos = 0;

	public JoinTaskRecordReader(FileSplit fileSplit, JobConf job)
			throws IOException {
		
		//ʹ�ô�����Ҫ���õ�����
		//��Ҫע�������������һ�������ڴ����޷����?����������inֻ��˳���?
		//����ֻ�ܱ������ļ�¼���ڴ棬��ôҲ��ֻ��������ϴ�������
		this.leftTableName = job.get("left.table.name");
		this.rightTableName = job.get("right.table.name");
		
		this.fileSplit = fileSplit;
		Path file = fileSplit.getPath();
		FileSystem fs = file.getFileSystem(job);
		this.in = fs.open(file);

		IOUtils.skipFully(in, fileSplit.getStart());
		
		byte[] tableDataHeader = new byte[tableDataHeaderLen];
		IOUtils.readFully(in, tableDataHeader, 0, tableDataHeaderLen);
		long tableDataSize = BinaryUtils.getLong(tableDataHeader, 1);

		doSelection(job, tableDataSize, fileSplit.getStart() + tableDataHeaderLen);
		
		IOUtils.readFully(in, tableDataHeader, 0, tableDataHeaderLen);
		rightTableDataSize = BinaryUtils.getLong(tableDataHeader, 1);
	}
	
	//�˺���������ǹ��˿϶����������join����е�����е�joinkey������ʣ�����м�¼��;��������
	//1���ڷ�join���Ͻ��й��ˣ�
	//2��ʹ��������join���Ͻ��й��ˣ������joinkey��ĳ����Χ��
	//���·���ԭ���ߣ�ʵ�ֶ�joinkey�Ϸ�Χ�Ĺ���
	public void doSelection(JobConf job, long size, long dataOffset) throws IOException{
		key = Integer.parseInt(job.get(JoinTaskInputFormat.IDX_ATTR));
		keyType = Integer.parseInt(job.get(JoinTaskInputFormat.IDX_ATTR_TYPE));
		keySize = Integer.parseInt(job.get(JoinTaskInputFormat.IDX_ATTR_SIZE));
		
		TableObjectProxy proxy = null;
		try {
			proxy = new TableObjectProxy(leftTableName);
		} catch (Exception e1) {
			e1.printStackTrace();
		}
		
		//������λ��
		IOUtils.skipFully(in, size);
		
		leftJoinRecord = new HashMap<String, List<TableObjectProxy>>();
		IndexLookupWithLongOffsets e = new IndexLookupWithLongOffsets(key, keyType, keySize, leftTableName);
		if (size > 0) {
			IndexLookupWithLongOffsets.TableObjectProxyList records = e.extract(in, job
					.get(JoinTaskInputFormat.LOW_KEY), job
					.get(JoinTaskInputFormat.HIGH_KEY), dataOffset);
			if (records != null) {
				long selectedDataSize = 0;
				while (records.hasNext()) {
					TableObjectProxy r = records.next();
					String joinKey = null;
					try {
						joinKey = r.getStringAttribute(key);
					} catch (Exception e2) {
						e2.printStackTrace();
					}
					if (leftJoinRecord.containsKey(joinKey))
						leftJoinRecord.get(leftJoinRecord).add(r);
					else {
						List<TableObjectProxy> list = new ArrayList<TableObjectProxy>();
						list.add(r);
						leftJoinRecord.put(joinKey, list);
					}
					try {
						selectedDataSize += proxy.getSize();
					} catch (Exception e1) {
						e1.printStackTrace();
					}
				}
				System.out.println("Selected data size: " + selectedDataSize);
			}
			records.skipRest();
		}
		IOUtils.skipFully(in, e.indexAndHeaderSize);
	}

	@Override
	public Text createKey() {
		return new Text();
	}

	@Override
	public Text createValue() {
		return new Text();
	}

	@Override
	public long getPos() throws IOException {
		return offset;
	}

	@Override
	public float getProgress() throws IOException {
		return in.getPos() / (float) fileSplit.getLength();
	}

	@Override
	public boolean next(Text key, Text value) throws IOException {
		
		if (joinKeyRecord != null) {
			
			TableObjectProxy r = joinKeyRecord.get(leftJoinRecordPos);
			
			try {
				key.set(r.getStringAttribute(this.key));
				value.set(r.getStringAttributes(null) + currentRightTableObjectProxy.getStringAttributes(null));
			} catch (Exception e1) {
				e1.printStackTrace();
			}
			
			if (leftJoinRecordPos < (joinKeyRecord.size() - 1))
				leftJoinRecordPos++;
			else
				joinKeyRecord = null;;
			return true;
		}
		
		while (offset < rightTableDataSize) {
			TableObjectProxy currentRightTableObjectProxy;
			try {
				currentRightTableObjectProxy = new TableObjectProxy(rightTableName);			
				int recordLen = currentRightTableObjectProxy.getSize();
				
				if (offset > rightTableDataSize)
					return false;
				
				byte[] data = new byte[recordLen];
				IOUtils.readFully(in, data, 0, recordLen);
				offset += recordLen;
				currentRightTableObjectProxy.setRecord(data);
				String joinKey = currentRightTableObjectProxy.getStringAttribute(this.key);

				if (leftJoinRecord.containsKey(joinKey)) {
					joinKeyRecord  = leftJoinRecord.get(joinKey);
					leftJoinRecordPos = 0;
					TableObjectProxy r = joinKeyRecord.get(leftJoinRecordPos);
					
					key.set(r.getStringAttribute(this.key));					
					value.set(r.getStringAttributes(null) + currentRightTableObjectProxy.getStringAttributes( null));
					
					if (leftJoinRecordPos < (joinKeyRecord.size() - 1))
						leftJoinRecordPos++;
					else
						joinKeyRecord = null;;
					
					return true;
				}
			} catch (Exception e) {
				e.printStackTrace();
			}
		}
		return false;
	}

	@Override
	public void close() throws IOException {
		IOUtils.closeStream(in);
	}
}
